// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Author: wsfuyibing <682805@qq.com>
// Date: 2024-12-05

package src

import (
	"context"
	"gitee.com/go-libs/log"
	"gitee.com/go-libs/process"
	"gitee.com/gomq/sr"
	amqp "github.com/rabbitmq/amqp091-go"
)

// Operate rabbitmq server.
type remoting struct {
	manager *manager

	name    string
	process process.Process
}

// Destroy
// remote settings, Unbind queues on the exchanges, Delete queues, Delete
// exchanges.
func (o *remoting) Destroy(ctx context.Context, subscription sr.Subscription) (err error) {
	var (
		ch   *amqp.Channel
		conn *amqp.Connection
	)

	log.Infofc(ctx, `[servlet=rabbitmq][remoting] destroy begin: subscription-id="%d"`, subscription.GetId())
	defer log.Infofc(ctx, `[servlet=rabbitmq][remoting] destroy finish`)

	// Clean when destroyed.
	defer func() {
		if r := recover(); r != nil {
			log.Fatalfc(ctx, `[servlet=rabbitmq][remoting] destroying caused an panic: %v`, r)
		}

		o.manager.client.CloseChannel(ctx, ch)
		o.manager.client.CloseConnection(ctx, conn)
	}()

	// Open connection.
	if conn, err = o.manager.client.OpenConnection(ctx); err != nil {
		return
	}

	// Open channel.
	if ch, err = o.manager.client.OpenChannel(ctx, conn); err != nil {
		return
	}

	// Open subscription settings mapper.
	exchange, _, _ := o.manager.client.GenerateExchange(subscription.GetId(), subscription.GetTopic())
	exchanges, queues, bindings := o.manager.client.GenerateMapping(subscription)

	// Delete bindings.
	for _, item := range bindings {
		if err = o.manager.client.DeleteBinding(ctx, ch, item.Queue, item.Topic, subscription.GetTag()); err != nil {
			return
		}
	}

	// Delete queues.
	for _, item := range queues {
		for key := range item {
			if err = o.manager.client.DeleteQueue(ctx, ch, key); err != nil {
				return
			}
		}
	}

	// Delete exchanges.
	for _, item := range exchanges {
		if item == exchange {
			continue
		}
		if err = o.manager.client.DeleteExchange(ctx, ch, item); err != nil {
			return
		}
	}
	return
}

// Initialize
// remote settings, Declare exchange as topic, Declare queues, Bind queues on
// exchanges.
func (o *remoting) Initialize(ctx context.Context, subscription sr.Subscription) (err error) {
	var (
		ch   *amqp.Channel
		conn *amqp.Connection
	)

	log.Infofc(ctx, `[servlet=rabbitmq][remoting] initialize begin: subscription-id="%d"`, subscription.GetId())
	defer log.Infofc(ctx, `[servlet=rabbitmq][remoting] initialize finish`)

	// Clean when initialized.
	defer func() {
		if r := recover(); r != nil {
			log.Fatalfc(ctx, `[servlet=rabbitmq][remoting] initializing caused an panic: %v`, r)
		}

		o.manager.client.CloseChannel(ctx, ch)
		o.manager.client.CloseConnection(ctx, conn)
	}()

	// Open connection.
	if conn, err = o.manager.client.OpenConnection(ctx); err != nil {
		return
	}

	// Open channel.
	if ch, err = o.manager.client.OpenChannel(ctx, conn); err != nil {
		return
	}

	// Open subscription settings mapper.
	exchanges, queues, bindings := o.manager.client.GenerateMapping(subscription)

	// Declare exchanges.
	for _, item := range exchanges {
		if err = o.manager.client.DeclareExchange(ctx, ch, item); err != nil {
			return
		}
	}

	// Declare queues.
	for _, item := range queues {
		for key, arg := range item {
			if err = o.manager.client.DeclareQueue(ctx, ch, key, arg); err != nil {
				return
			}
		}
	}

	// Declare bindings.
	for _, item := range bindings {
		if err = o.manager.client.DeclareBinding(ctx, ch, item.Queue, item.Topic, subscription.GetTag()); err != nil {
			return
		}
	}
	return
}

// Process
// returns the virtual process like os.
func (o *remoting) Process() process.Process {
	return o.process
}

// + Process

func (o *remoting) Run(ctx context.Context) (err error) {
	for {
		select {
		case <-ctx.Done():
			return
		}
	}
}

// + Access

func (o *remoting) init() *remoting {
	o.name = "rabbitmq-remoting"
	o.process = process.New(o.name, o)
	return o
}
